package net.bwie.realtime.jtp.ods.job;

import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.common.utils.MysqlCdcUtil;
import net.bwie.realtime.jtp.ods.Function.CdcDataEtlMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class JtpCdcMallDataOdsJob {

    public static void main(String[] args) throws Exception {
        //1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.enableCheckpointing(5000L);


        //2 数据源 - source
        DataStream<String>  cdcStream =cdcMysqlSource(env);
//        cdcStream.print("aaa");


        //3 数据转换 - transformation
         DataStream<String>  etlStream = cdcStream.map(new CdcDataEtlMapFunction());




        //4 数据输出 - sink
        KafkaUtil.producerKafka(etlStream,"topic-db");



        // 5 触发执行

        env.execute("JtpCdcMallDataOdsJob");



    }

    private static DataStream<String> cdcMysqlSource(StreamExecutionEnvironment env) {

        //todo s1 全量+增量： 基础维度表数据

        String[]  dimList = new String[]{
                "base_dic","base_province","base_region","base_category1","base_category2","base_category3",
                "activity_info","activity_rule","base_trademark","coupon_info","sku_info","spu_info","user_info"
        };

        DataStream<String> cdcDimStream = MysqlCdcUtil.cdcMysqlInitial(env, "jtp_mall", dimList);

        //todo s2 增量：业务事实表数据
        String[]  factList = new String[]{
                "favor_info","order_info","cart_info","order_detail","order_detail_activity","order_detail_coupon",
                "order_status_log","payment_info","order_refund_info","refund_payment","comment_info","coupon_use"
        };

        DataStream<String> cdcFactStream = MysqlCdcUtil.cdcMysqlEarliest(env, "jtp_mall", factList);


        DataStream<String> cdcStream = cdcDimStream.union(cdcFactStream);


        return cdcStream;
    }


}
